package com.llx.datastream

import java.text.SimpleDateFormat

import com.sun.jmx.snmp.Timestamp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.RichSourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.functions.{AscendingTimestampExtractor, AssignerWithPeriodicWatermarks, KeyedProcessFunction}
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.table.shaded.org.joda.time.DateTime
import org.apache.flink.util.Collector

import scala.util.Random

/**
  * dataStream：
  * keyedStream：
  * 主要内容：
  * 1、实现RichSourceFunction 的run方法，定义DataStream 数据源
  *
  * 2、addSource
  *
  * 3、修改默认的ProcessingTime处理方式，变成 EventTime 模式进行处理
  *   env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  *
  * 4、获得业务时间，以及生成 Watermark
  *     itemStream.assignTimestampsAndWatermarks
  *
  * 5、添加 timeWindow窗口，调用聚合函数统计 aggregate(AggregateFunction(聚合函数),WindowFunction(窗口返回聚合后的dataStream))
  *      with ListCheckpointed可以实现 snapshotState存储  restoreState 回复  apply 处理
  *      并开启快照：env.getCheckpointConfig.setCheckpointInterval(6000)
  *
  *
  * 6、对统计厚的 dataStream.process(KeyedProcessFunction)
  *   1、open 初始化定义处理的状态 元素列表
  *       getRuntimeContext.getListState[ItemViewCount](itemsStateDesc)
  *   2、processElement(处理元素) ----> registerEventTimeTimer（再次注册定时器）   ---->onTimer（触发定时器）
  *      注册定时器
  *      context.timerService.registerEventTimeTimer(i.windowEnd + 1)
  *
  * 7、
  *
  *
  *
  */


object DataStreamExample {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    socket(env)
//    keyByTest(env)
//    水印兼容延迟(env)
//    窗口统计点击量(env)

    /**
      * TopN计算最热门商品 依赖 窗口统计点击量() 统计出来的结果再次执行窗口函数，
      * 所以把 窗口统计点击量()代码变形后 添加进去
      *
      * 窗口统计点击量()  现在依赖于tupe方式实现的没有依赖于  TopN计算最热门商品() item...对象
      */
    TopN计算最热门商品(env)

  }


  /**dst1：DataStream
    * dst1.assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks[dst1类型])
    * @param env
    */
  def 水印兼容延迟(env : StreamExecutionEnvironment): Unit ={
    env.setParallelism(1)//注意控制并发数
    //置为使用EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val source = env.socketTextStream("localhost", 9999)

    val dst1: DataStream[SalePrice] = source.map(value => {
      val columns = value.split(",")
      SalePrice(columns(0).toLong, columns(1), columns(2), columns(3).toDouble)
    })
    //todo 水印时间 assignTimestampsAndWatermarks
    val timestamps_data = dst1.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[SalePrice]{

      var currentMaxTimestamp:Long = 0
      val maxOutOfOrderness = 2000L //最大允许的乱序时间是2s
      var wm : Watermark = null
      val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")

      /**
        * 获取接收元素的最大时间---当前接收元素的最大时间  ---也是当前最大时间
        * @param element
        * @param previousElementTimestamp
        * @return
        */
      override def extractTimestamp(element: SalePrice, previousElementTimestamp: Long): Long = {
        val timestamp = element.time //得到元素的接收时间
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)  //
        currentMaxTimestamp
      }

      /**
        * 获取水位线：水位线时间之前的数据都已经接收完成
        *
        * @return
        */
      override def getCurrentWatermark: Watermark = {
        wm = new Watermark(currentMaxTimestamp - maxOutOfOrderness)
        wm
      }
    })

    val data: KeyedStream[SalePrice, String] = timestamps_data.keyBy(line => line.productName)
    val window_data: WindowedStream[SalePrice, String, TimeWindow] = data.timeWindow(Time.seconds(3))
    val apply: DataStream[SalePrice] = window_data.apply(new MyWindowFunc)
    apply.print()
    env.execute()
}

  case class SalePrice(time: Long, boosName: String, productName: String, price: Double)

  /**
    * [SalePrice , SalePrice , String, TimeWindow] 代表apply方法的参数类型[input类型，out类型，key类型，window]
    */
  class MyWindowFunc extends WindowFunction[SalePrice , SalePrice , String, TimeWindow]{
    override def apply(key: String, window: TimeWindow, input: Iterable[SalePrice], out: Collector[SalePrice]): Unit = {
      val seq = input.toArray
      val take: Array[SalePrice] = seq.sortBy(line => line.price).reverse.take(1)
      for(info <- take){
        out.collect(info)
      }
    }
  }


  /**
    *
    * @param env
    */
  def socket(env : StreamExecutionEnvironment): Unit ={
    //设置checkpoint模式
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //checkpoint做小间隔500ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    //设置checkpoint间隔
    env.getCheckpointConfig.setCheckpointInterval(6000)
    //checkpoint操作期间最长1分钟，超过会放弃checkpoint
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    //如果任务的检查点发生错误，检查点就会被拒绝
    env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
    //同时只允许一个检查点在进行中
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

//////////////////////////////////////////////////////////////////////////////////
    /** dataStream 的split函数与select函数
      *
      */

    val elementsStream: DataStream[Int] = env.fromElements(1,2,3,4,5,6)
    env.setParallelism(1) //设置分区为1
    //数据分流
    val split_data_stream = elementsStream.split((num: Int) => {
      (num % 2) match {
        case 0 => List("even")
        case 1 => List("odd")
      }
    })

    //获取分流后的数据 dataStream通过split方式分流后，通select函数，获取其中分流的数据（list）
    val select_stream: DataStream[Int] = split_data_stream.select("even")
    select_stream.print()
    env.execute()

//////////////////////////////////////////////////////////////////////////////////

    val textStream = env.socketTextStream("localhost" , 9999)

    val lowerData: DataStream[String] = textStream
        .flatMap(_.toLowerCase().split("\\W+"))

    val nonEmpty_data: DataStream[String] = lowerData
        .filter(line => line.nonEmpty)

    val mapData: DataStream[(String, Int)] = nonEmpty_data
        .map(line => (line , 1))

    val sum: DataStream[(String, Int)] = mapData
        .keyBy(0)
        .sum(1)
    sum.print()

    //window keyBy相当于groupBy

    val windowCounts = mapData.keyBy(_._1)

      /**tumbling-time-window (无重叠数据)
        * 执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5秒
        * 也就是说，每5秒钟统计一次，在这过去的5秒钟内，各个路口通过红绿灯汽车的数量。
        */
//       .timeWindow(Time.seconds(2))

      /**tumbling-count-window (无重叠数据)
        * .执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5
        * 按照key进行收集，对应的key出现的次数达到5次作为一个结果
        */
//            .countWindow(5)

      /**sliding-count-window (有重叠数据)
        *.执行统计操作，每个sensorId一个sliding窗口，窗口大小3条数据,窗口滑动为3条数据
        * /也就是说，每个路口分别统计，收到关于它的3条消息时统计在最近5条消息中，各自路口通过的汽车数量
        */
//      .countWindow(5, 3)

      /**sliding-time-window (有重叠数据)
        * .执行统计操作，每个sensorId一个sliding窗口，窗口时间10秒,滑动时间5秒
        * 也就是说，每5秒钟统计一次，在这过去的10秒钟内，各个路口通过红绿灯汽车的数量。
        */
      .timeWindow(Time.seconds(2), Time.seconds(2))   //窗口大小  滑动间隔

      /**所以，我们需要考虑到网络延迟的状况，那么代码中就需要添加水印操作：
        *
        */
      .sum("count")

    windowCounts.print()
      .setParallelism(1)

    env.execute("start streaming window wordCount")

  }

  def 窗口统计点击量(env : StreamExecutionEnvironment): Unit ={

//    val textStream = env.socketTextStream("localhost" , 9999)
    val textStream =  env.addSource(new SEventSourceWithChk).setParallelism(1)

    val lowerData: DataStream[String] = textStream
      .flatMap(_.split(","))

    val nonEmpty_data: DataStream[String] = lowerData
      .filter(line => line.nonEmpty)

    val mapData: DataStream[(String, Int)] = nonEmpty_data
      .map(line => (line , 1))
    val windowCounts = mapData.keyBy(0)
      .timeWindow(Time.seconds(2), Time.seconds(2))   //窗口大小  滑动间隔


      /** 2、方法表示1
        * mapData.keyBy(_._1) keybY明确了哪个字段（数据类型）
        * aggregate（统计函数，window输出函数）
        *
        * .aggregate(AggregateFunction af, WindowFunction wf) 做增量的聚合操作，它能使用
        * AggregateFunction 提前聚合掉数据，减少 state 的存储压力。较之.apply(WindowFunction wf)
        * 会将窗口中的数据都存储下来，最后一起计算要高效地多。aggregate()方法的:
        * 第一个参数:
        *   用于这里的CountAgg 实现了AggregateFunction 接口，功能是统计窗口中的条数，即遇到一条数据就加一。
        *
        * 第二个参数WindowFunction
        *     将每个 key 每个窗口聚合后的结果带上其他信息进行输出。
        * 我们这里实现的WindowResultFunction将主键商品ID，窗口，点击量封装成了ItemViewCount 进行输出。
        *
        */

      /**2、方法表示2
        * mapData.keyBy(0) 用数字表示keyby，不知道数据类型，在aggregate（）的第二给参数WindowFunction 窗口的主键是tuple类型
        *
        *
        */
      .aggregate(
            new AggregateFunction[(String,Int),Int,Int] {
              //添加元素行为
              override def add(in: (String, Int), acc: Int): Int = {
                acc+1

              }
              //初始值
              override def createAccumulator(): Int = {
                println("createAccumulator---->")
                0
              }
              //获得结果
              override def getResult(acc: Int): Int = {
                println("getResult---->"+acc)
                acc
              }
              //        合并
              override def merge(acc: Int, acc1: Int): Int = {
                println("merge--------->"+acc +"-------->"+ acc1)
                acc + acc1
              }
            },
      new WindowFunction[Int,(String,Int),Tuple,TimeWindow]{
                /**类似于apply
                  *
                  * @param key
                  * @param window
                  * @param input
                  * @param out
                  */
                                       //窗口的主键       窗口              聚合函数的结果，即count 值  输出类型为Collector
                override def apply(key: Tuple, window: TimeWindow, input: Iterable[Int], out: Collector[(String,Int)]): Unit = {
                  val i = input.iterator.next()
                  println("apply--key---------------->" +key + "------>" + i)
                                         println("iiiiiiiiiii" + key.getField(0))
                  out.collect((key.getField(0),i))
                  println("------------------")
                }
              }
    )
    windowCounts.print()
    env.execute("start streaming window wordCount")

  }


  /**通过每个窗口统计结果，再次进行topN  TopN计算最热门商品   process(KeyedProcessFunction)
    *
    * 为了统计每个窗口下最热门的商品，我们需要再次按窗口进行分组，这里根据ItemViewCount
    * 中的windowEnd 进行keyBy()操作。然后使用 ProcessFunction 实现一个自定义的 TopN 函数
    * TopNHotItems 来计算点击量排名前3 名的商品，并将排名结果格式化成字符串，便于后续输出。
    * @param env
    */
  def TopN计算最热门商品(env : StreamExecutionEnvironment): Unit ={
   var ret = 窗口统计总商品返回窗口统计结果(env).keyBy("windowEnd")

         /**
           * ProcessFunction 是 Flink 提供的一个 low-level API，用于实现更高级的功能。
           * 它主要提供了
           * 定时器 timer 的功能（支持EventTime 或ProcessingTime）。
           * 本案例中我们将利用 timer 来判断何时收齐了某个 window 下所有商品的点击量数据。
           * 由于 Watermark 的进度是全局的，
           * 在processElement 方法中，每当收到一条数据（ItemViewCount），我们就注册一个 windowEnd+1
           *  的定时器（Flink 框架会自动忽略同一时间的重复注册）。windowEnd+1 的定时器被触发时，意
           *  味着收到了windowEnd+1 的 Watermark，即收齐了该windowEnd 下的所有商品窗口统计值。我
           *  们在onTimer()中处理将收集的所有商品及点击量进行排序，选出 TopN，并将排名信息格式化
           *  成字符串后进行输出。

           *  这里我们还使用了ListState<ItemViewCount>来存储收到的每条ItemViewCount 消息，保证在
           *  发生故障时，状态数据的不丢失和一致性。ListState 是 Flink 提供的类似 Java List 接口的State API，
           *  它集成了框架的 checkpoint 机制，自动做到了 exactly-once 的语义保证。
           *    org.apache.flink.api.java.tuple.Tuple
           *
           *
           *    CoProcessFunction----->类似join 链接两个流
           *    例子：
           *    https://www.jianshu.com/p/e6297fac67cb
           */

      .process(keyedProcessFunction = new KeyedProcessFunction[Tuple,ItemViewCount,String] {
//          import org.apache.flink.api.common.state.ListStateDescriptor
          var topiSize:Int = 2
     /**  通过itemState 来把所有的状态搜集起来进行，统计 及处理
       * itemState:ListState[ItemViewCount]
       */
          var itemState:ListState[ItemViewCount] = null

     /**
       * ListStateDescriptor : 状态描述器
       * getRuntimeContext.getListState :ListState --> 通过状态描述器，获得所有状态 仅仅当前process中的共享
       * @param parameters
       */
          override def open(parameters: Configuration): Unit = {
            super.open(parameters)
            val itemsStateDesc = new ListStateDescriptor[ItemViewCount]("itemState-state", classOf[ItemViewCount])
            itemState = getRuntimeContext.getListState[ItemViewCount](itemsStateDesc)
          }

     /**
       * 处理每一个元素 并搜集所有状态放到 itemState   此处还可以修改元素的状态
       * 处理函数
       *  是Flink提供的最具有表现力的接口。
       * Flink 提供的ProcessFuntion{低层次的流处理操作}是用来处理来自于一个或者两个输入流或者一段时间窗内的聚合的独立event。
       * ProcessFunction提供了对时间和状态的细粒度控制。一个ProcessFunction可以任意的修改它的状态，
       * 并且也可以注册定时器在将来触发一个回调函数。因此，
       * ProcessFunction可以根据需要为有很多有状态的事件驱动的应用实现复杂的单事件业务逻辑。
       */
          override def processElement(i: ItemViewCount, context: KeyedProcessFunction[Tuple, ItemViewCount, String]#Context, collector: Collector[String]): Unit = {
            // 每条数据都保存到状态中
            itemState.add(i)
            // 注册windowEnd+1 的EventTime Timer, 当触发时，说明收齐了属于windowEnd 窗口的所
//            有商品数据
            /**
              * 注册定时器：
              * 注册定时器在将来触发一个回调函数
              * context.timerService.registerEventTimeTimer
              */
            context.timerService.registerEventTimeTimer(i.windowEnd + 1)

            println("================>" + i)
          }

          /**scala 中使用java代码编程
            * 1、   <>   ---->   [ItemViewCount]
            * 2、   导入  import scala.collection.JavaConversions._
            * 3、   for循环 带下标的
            *         for (i <- 0 to topiSize)
            * 4、   tuple -----> ()
            * 5、   class   ---->   classOf()
            *
            *
            */

     /**
       * 触发后，处理上面 注册的定时器
       * @param timestamp
       * @param ctx
       * @param out
       */
     override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
//            super.onTimer(timestamp, ctx, out)
            println("进入 onTimer")
            import java.util
            import java.util.Comparator
            // 获取收到的所有商品点击量// 获取收到的所有商品点击量

            val allItems = new util.ArrayList[ItemViewCount]
            import scala.collection.JavaConversions._
            for (item <- itemState.get) {
              allItems.add(item)
            }
            // 提前清除状态中的数据，释放空间
            itemState.clear
            // 按照点击量从大到小排序
//            allItems.sortBy((e)=>{
//              -e.viewCount
//            })
            allItems.sort(new Comparator[ItemViewCount]() {
              override def compare(o1: ItemViewCount, o2: ItemViewCount): Int = (o2.viewCount - o1.viewCount).asInstanceOf[Int]
            })
            // 将排名信息格式化成String, 便于打印
            val result = new StringBuilder
            result.append("====================================\n")
            result.append("时间: ").append(new Timestamp(timestamp - 1)).append("\n")
            for (i <- 1 to topiSize) {
              if(topiSize <= allItems.length){
                val currentItem  = allItems.get(i)
                // No1: 商品ID=12224 浏览量=2413
                result.append("No").append(i).append(":")
                  .append(" 商品ID=").append(currentItem.itemId)
                  .append(" 浏览量=").append(currentItem.viewCount)
                  .append("\n")
              }
            }
            result.append("====================================\n\n")
            out.collect(result.toString())
            println("result----->" + result)
          }
        } )
    ret.print()
    env.execute("Hot Items Job")


  }

  /**
    * 通过自定义源，通过自定义递增的timestemp  设置watermark 并同过 keyby 与timewindow 函数 统计时间窗口的数据
    * @param env
    * @return
    */
  def 窗口统计总商品返回窗口统计结果(env: StreamExecutionEnvironment): DataStream[ItemViewCount] ={
  /**
    * 通过inportFormat  创建dataStream
    * eg:
    * PojoCsvInputFormat<>
    *
    */

//  val value = new PojoCsvInputFormat[UserBehavior]()
//env.createInput()

      env.setParallelism(1)
  val itemStream =  env.addSource(new itemSource)
  import org.apache.flink.streaming.api.TimeCharacteristic

  /** 模拟实验，自定义窗口时间需要定义一下两件事没否则  onTimer 函数无法被调用（无法触发事件）
    * 第一件是：
    * 告诉 Flink 我们现在按照 EventTime 模式进行处理，Flink 默认使用 ProcessingTime处理，所以我们要显式设置下。
    */
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

  /**
    * 第二件事情:   定如何获得业务时间，以及生成 Watermark
    * 是指定如何获得业务时间，以及生成 Watermark。Watermark 是用来追踪业务事件
    * 的概念，可以理解成 EventTime 世界中的时钟，用来指示当前处理到什么时刻的数据了。由于
    * 我们的数据源的数据已经经过整理，没有乱序，即事件的时间戳是单调递增的，所以可以将每
    * 条数据的业务时间就当做 Watermark。这里我们用AscendingTimestampExtractor 来实现时间戳的抽取和 Watermark 的生成。
    */

  /**自定义waterMark 替代方案
    *   AssignerWithPeriodicWatermarks {周期性的分类器  通过自己的timestamp}和
    *   AssignerWithPunctuatedWatermarks { 发射自己的watermark 通过实现checkAndGetNextWatermark函数，返回下一个 Watermark}
    *
    * flink已实现的分类器：
    *   递增时间戳分配器：AscendingTimestampExtractor
    *   允许固定数量延迟的分配器：BoundedOutOfOrdernessTimestampExtractor  (Time.seconds(10))( _.getCreationTime )
    *
    */
  val windowCounts = itemStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor[UserBehavior] {
    override def extractAscendingTimestamp(t: UserBehavior): Long = {
      // 原始数据单位秒，将其转成毫秒
      t.timestamp
    }
  }).keyBy("itemId")
    .timeWindow(Time.seconds(5), Time.seconds(2)) //窗口大小  滑动间隔
    .aggregate(
    //实现聚合
    new AggregateFunction[UserBehavior, Int, Int] {
      //添加元素行为
      override def add(in: UserBehavior, acc: Int): Int = {
        acc + 1

      }

      //初始值
      override def createAccumulator(): Int = {
        //        println("createAccumulator---->")
        0
      }

      //获得结果
      override def getResult(acc: Int): Int = {
        //        println("getResult---->"+acc)
        acc
      }

      //        合并
      override def merge(acc: Int, acc1: Int): Int = {
        //        println("merge--------->"+acc +"-------->"+ acc1)
        acc + acc1
      }
    },
    // 实现窗口操作：窗口内的状态搜集：
    new WindowFunction[Int, ItemViewCount, Tuple, TimeWindow] {
      /** 类似于apply
        *
        * @param key
        * @param window
        * @param input
        * @param out
        */
      //窗口的主键       窗口              聚合函数的结果，即count 值  输出类型为Collector
      override def apply(key: Tuple, window: TimeWindow, input: Iterable[Int], out: Collector[ItemViewCount]): Unit = {
        val i = input.iterator.next()
        println("apply--key---------------->" + key + "------>" + i)
        out.collect(ItemViewCount(key.getField(0), window.getEnd, input.iterator.next()))

      }
    }
  )
  windowCounts
}

  /**
    * 定义用到的实体
    * @param itemId
    * @param windowEnd
    * @param viewCount
    */
  case class ItemViewCount(itemId:Long,windowEnd:Long ,viewCount:Long)
  //categoryId:商品类目ID  behavior:用户行为, 包括("pv", "buy", "cart", "fav")  timestamp:行为发生的时间戳，单位秒
  case class UserBehavior(userId:Long,itemId:Long,categoryId:Int,behavior:String,timestamp:Long)
  /**
    * ，模拟数据流
    */
  /**
    * ，模拟数据流
    */
  class itemSource extends RichSourceFunction[UserBehavior]{
    private var count = 0L
    private var isRunning = true
    private val alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWZYX0987654321"
    // 任务取消时调用
    override def cancel(): Unit = {
      isRunning = false
    }
    // source算子的逻辑，即:每秒钟向流图中注入10000个元组
    override def run(sourceContext: SourceContext[UserBehavior]): Unit = {
      while(isRunning) {
        val time =   new DateTime().getMillis-Random.nextInt(1000)
        for (i <- 0 until 10) {
          sourceContext.collect(UserBehavior(Random.nextInt(5),Random.nextInt(3),Random.nextInt(3),"pv",time ))
          count += 1L
        }
        Thread.sleep(1000)
      }
    }
  }


  class SEventSourceWithChk extends RichSourceFunction[String]{
    private var count = 0L
    private var isRunning = true
    private val alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWZYX0987654321"
    // 任务取消时调用
    override def cancel(): Unit = {
      isRunning = false
    }
    // source算子的逻辑，即:每秒钟向流图中注入10000个元组
    override def run(sourceContext: SourceContext[String]): Unit = {
      while(isRunning) {
        for (i <- 0 until 10) {
          sourceContext.collect("a,b,b,c,c,c")
          count += 1L
        }
        Thread.sleep(1000)
      }
    }
  }


  class IntSocket extends RichSourceFunction[String]{
    private var count = 0L
    private var isRunning = true
    private val alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWZYX0987654321"
    // 任务取消时调用
    override def cancel(): Unit = {
      isRunning = false
    }
    // source算子的逻辑，即:每秒钟向流图中注入10000个元组
    override def run(sourceContext: SourceContext[String]): Unit = {
      while(isRunning) {
        for (i <- 0 until 10) {
          sourceContext.collect("1,2,3,1,2,1")
          count += 1L
        }
        Thread.sleep(1000)
      }
    }
  }
}

